package io.github.wanggit.antrpc.client.zk.listener;

import io.github.wanggit.antrpc.client.zk.IZkClient;
import io.github.wanggit.antrpc.client.zk.lb.ILoadBalancerContainer;
import io.github.wanggit.antrpc.commons.bean.SubscribeNode;
import io.github.wanggit.antrpc.commons.config.IConfiguration;
import io.github.wanggit.antrpc.commons.constants.ConstantValues;
import io.github.wanggit.antrpc.commons.utils.ApplicationNameUtil;
import io.github.wanggit.antrpc.commons.utils.JsonUtils;
import io.github.wanggit.antrpc.commons.zookeeper.ZkNodeType;
import lombok.extern.slf4j.Slf4j;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.cache.ChildData;
import org.apache.curator.framework.recipes.cache.TreeCache;
import org.apache.curator.framework.recipes.cache.TreeCacheEvent;

import java.util.Objects;

@Slf4j
public class SubscribeNodeChangeListener implements IListener {

    private final IZkClient zkClient;
    private final ILoadBalancerContainer loadBalancerContainer;
    private final IConfiguration configuration;

    public SubscribeNodeChangeListener(
            IZkClient zkClient,
            ILoadBalancerContainer loadBalancerContainer,
            IConfiguration configuration) {
        this.zkClient = zkClient;
        this.loadBalancerContainer = loadBalancerContainer;
        this.configuration = configuration;
    }

    @Override
    public void listen() {
        internalListen(zkClient.getCurator());
    }

    /**
     *
     *
     * <pre>
     *     <[CDATA[
     *     /__rpc_sub__/log_report_kafka_service_1@172.16.101.19:6062/io.github.wanggit.antrpc.demo.api.HelloService
     *     {
     *          "host" : "log_report_kafka_service_1@172.16.101.19:6062",
     *          "className" : "io.github.wanggit.antrpc.demo.api.HelloService",
     *          "ts" : 1592965776836,
     *          "methodMap" : null
     *     }
     *
     *     methodMap => key -> RegisterBeanMethod.toString, value -> registerBeanMethod
     *     ]]>
     * </pre>
     *
     * 监听订阅服务的变化，此变化可能带来限流信息的变化
     *
     * @param curatorFramework curator
     */
    private void internalListen(CuratorFramework curatorFramework) {
        TreeCache treeCache =
                new TreeCache(curatorFramework, "/" + ConstantValues.ZK_ROOT_SUBSCRIBE_NODE_NAME);
        treeCache
                .getListenable()
                .addListener(
                        (client, event) -> {
                            if (Objects.equals(event.getType(), TreeCacheEvent.Type.NODE_UPDATED)
                                    || Objects.equals(
                                            event.getType(), TreeCacheEvent.Type.NODE_ADDED)) {
                                ChildData childData = event.getData();
                                if (null != childData) {
                                    String path = childData.getPath();
                                    ZkNodeType.SubscribeNodeType subscribeNodeType =
                                            ZkNodeType.getSubscribeNodeType(path);
                                    if (Objects.equals(
                                            subscribeNodeType,
                                            ZkNodeType.SubscribeNodeType.INTERFACE)) {
                                        SubscribeNode subscribeNode =
                                                JsonUtils.fromJson(
                                                        childData.getData(), SubscribeNode.class);
                                        // 只处理本节点的消费端限流信息
                                        String applicationName =
                                                ApplicationNameUtil.getApplicationName(
                                                        configuration.getExposeIp(),
                                                        configuration.getApplicationName(),
                                                        configuration.getPort());
                                        if (Objects.equals(
                                                applicationName, subscribeNode.getHost())) {
                                            loadBalancerContainer.updateHostRateLimit(
                                                    subscribeNode);
                                        }
                                    }
                                }
                            }
                        });
        try {
            treeCache.start();
        } catch (Exception e) {
            String errorMessage = "Subscribe Node ChangeTreeCache failed to start.";
            if (log.isErrorEnabled()) {
                log.error(errorMessage, e);
            }
            throw new ZkListenerException(errorMessage, e);
        }
    }
}
